package com.yuan.yblog.portal.listener;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.yuan.yblog.portal.config.RabbitmqConfig;
import com.yuan.yblog.portal.pojo.DTO.MysqlMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @author hepeiyuan on 2024/5/30 22:31
 * @description canal监听器，监听mysql日志变化
 */
@Service
@Slf4j
public class CanalListener {

    @Autowired
    CanalConnector connector;

    @Autowired
    RabbitTemplate rabbitTemplate;


    /**
     * 批操作数目
     */
    private final static int BATCH_SIZE = 1000;

    /**
     *  监听间隔，单位：毫秒
     */
    private final static long WAIT_TIME = 2000;

    /**
     * canal监听
     */
    public void canalListen() throws Exception {
        try {
            //打开连接
            connector.connect();
            //订阅数据库表,全部表
            connector.subscribe(".*\\..*");
            //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
            connector.rollback();
            log.info("--canal监听启动--");
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(BATCH_SIZE);
                //获取批量ID
                long batchId = message.getId();
                //获取批量的数量
                int size = message.getEntries().size();
                //如果没有数据
                if (batchId == -1 || size == 0) {
                    try {
                        //线程休眠
                        Thread.sleep(WAIT_TIME);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    //如果数据发生变化，投递“同步数据”的消息
                    deliverSynchronousMessage(message.getEntries());
                }
                //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
                connector.ack(batchId);
            }
        } finally {
            connector.disconnect();
        }
    }

    /**
     * 投递 mysql-es 数据同步消息
     * @param entries 这次操SQL作记录
     */
    private void deliverSynchronousMessage(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            String tableName = entry.getHeader().getTableName();
            // 开启/关闭事务的实体类型，跳过
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            //RowChange对象，包含了一行数据变化的所有特征，比如isDDL 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry, e);
            }

            //获取操作类型：insert/update/delete类型
            CanalEntry.EventType eventType = rowChange.getEventType();

            //判断是否是DDL语句，若是则忽略
            if (rowChange.getIsDdl()) {
                continue;
            }

            //获取RowChange对象里的每一行数据
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.INSERT) {
                    // insert语句，取【操作之后】的数据
                    Long id = getRowId(rowData.getAfterColumnsList());
                    if (id != null) {
                        deliverMessage(new MysqlMessage(id, tableName, MysqlMessage.OperationType.INSERT));
                    }
                } else if (eventType == CanalEntry.EventType.DELETE) {
                    // delete语句，取【操作之前】的数据
                    Long id = getRowId(rowData.getBeforeColumnsList());
                    if (id != null) {
                        deliverMessage(new MysqlMessage(id, tableName, MysqlMessage.OperationType.DELETE));
                    }
                } else if (eventType == CanalEntry.EventType.UPDATE) {
                    // update的语句，取【操作之后】的数据
                    Long id = getRowId(rowData.getAfterColumnsList());
                    if (id != null) {
                        deliverMessage(new MysqlMessage(id, tableName, MysqlMessage.OperationType.UPDATE));
                    }
                }
            }
        }
    }

    /**
     * 获取当前sql语句的数据的主键（id）
     * @param columns 该行数据的各字段集合
     * @return 主键id值
     */
    private Long getRowId(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            if (column.getName().equals("id")) {
                return Long.valueOf(column.getValue());
            }
        }
        return null;
    }

    /**
     * 投递消息
     * @param message 消息对象
     */
    public void deliverMessage(MysqlMessage message) {
        rabbitTemplate.convertAndSend(
                RabbitmqConfig.EXCHANGE_REDIS_MYSQL_SYNC,
                RabbitmqConfig.ROUTING_REDIS_MYSQL_SYNC,
                message
        );
    }
}
